package cc.owoo.godpen.thread;

import cc.owoo.godpen.network.throughput.Throughput;
import cc.owoo.godpen.util.N;

import java.util.HashMap;
import java.util.LinkedList;

/**
 * 下载线程，这个线程会根据网速自动调整线程池大小，已保证最大化吃满带宽
 * Created by nimensei
 * 2022-03-31 上午 01:58
 **/
public class SyncDownloadThread {
    private int minThreadCount = 1;// 最小线程数
    private int maxThreadCount = Integer.MAX_VALUE;// 最大线程数
    private final SyncThreadPool pool;
    private final HashMap<Throughput, ThroughputRecord> throughputMap = new HashMap<>();// 流量计数器

    private double rate;// 当前速率
    private OnRateUpdate onRateUpdate;// 速率更新回调函数

    /**
     * 初始化线程池，初始线程数为3
     */
    public SyncDownloadThread() {
        this(3);
    }

    /**
     * 初始化线程池
     *
     * @param initThreadCount 初始线程数
     */
    public SyncDownloadThread(int initThreadCount) {
        pool = new SyncThreadPool(initThreadCount);
    }

    /**
     * 设置最小线程数
     *
     * @param minThreadCount 最小线程数
     */
    public void setMinThreadCount(int minThreadCount) {
        this.minThreadCount = minThreadCount;
    }

    /**
     * 获取最小线程数
     *
     * @return 最小线程数
     */
    public int getMinThreadCount() {
        return minThreadCount;
    }

    /**
     * 设置最大线程数
     *
     * @param maxThreadCount 最大线程数
     */
    public void setMaxThreadCount(int maxThreadCount) {
        this.maxThreadCount = maxThreadCount;
    }

    /**
     * 获取最大线程数
     *
     * @return 最大线程数
     */
    public int getMaxThreadCount() {
        return maxThreadCount;
    }

    /**
     * 获取当前正在运行的线程数
     *
     * @return 正在运行的线程数
     */
    public int getNowThreadCount() {
        return pool.getNowThreadCount();
    }

    /**
     * 添加流量计数器
     *
     * @param throughput 流量计数器
     */
    public boolean add(Throughput throughput) {
        if (throughput == null)
            throw new NullPointerException("流量计数器不能为空");
        synchronized (throughputMap) {
            if (throughputMap.containsKey(throughput))
                return false;
            ThroughputRecord value = new ThroughputRecord();
            throughputMap.put(throughput, value);
        }
        return true;
    }

    /**
     * 移除流量计数器
     *
     * @param throughput 流量计数器
     */
    public boolean remove(Throughput throughput) {
        if (throughput == null)
            throw new NullPointerException("流量计数器不能为空");
        synchronized (throughputMap) {
            return throughputMap.remove(throughput) != null;
        }
    }

    /**
     * 临时添加流量计数器
     *
     * @param throughput 流量计数器
     * @param runnable   执行回调函数（不能为空）
     */
    public void count(Throughput throughput, Runnable runnable) {
        if (runnable == null)
            throw new NullPointerException("执行回调函数不能为空");
        add(throughput);
        runnable.run();
        remove(throughput);
    }

    /**
     * 计数并获取当前吞吐量（发送与接收的速率），单位为：byte/秒
     *
     * @return 速率
     */
    public double getRate() {
        return rate;
    }

    /**
     * 设置速率
     *
     * @param rate 速率
     */
    private void setRate(double rate) {
        this.rate = rate;
        if (onRateUpdate != null)
            onRateUpdate.update(rate);
    }

    /**
     * 设置速率更新回调函数
     *
     * @param onRateUpdate 速率更新回调函数
     */
    public void setOnRateUpdate(OnRateUpdate onRateUpdate) {
        this.onRateUpdate = onRateUpdate;
    }

    /**
     * 执行一个线程回调函数
     *
     * @param runnable 线程回调函数
     */
    public void execute(Runnable runnable) {
        pool.execute(runnable);
        modulation.start();
    }

    /**
     * 等待结束
     */
    public void finish() {
        pool.awaitFinish();
        modulation.stop();
    }

    /**
     * 速率更新监听器
     */
    public interface OnRateUpdate {
        void update(double rate);
    }

    /**
     * 计算速率，单位：byte/毫秒
     *
     * @return 速率
     */
    private double calcRate(long timeMillis) {
        double speed = 0;
        synchronized (throughputMap) {
            for (var entry : throughputMap.entrySet()) {
                Throughput throughput = entry.getKey();
                ThroughputRecord record = entry.getValue();
                if (record.time == 0) {
                    record.record(throughput, timeMillis);
                    continue;
                }
                long value = record.value;
                long time = record.time;
                record.record(throughput, timeMillis);
                value = record.value - value;
                time = Math.max(1, record.time - time);
                if (value <= 0)
                    continue;
                speed += (double) value / (double) time;
            }
        }
        return speed;
    }

    /**
     * 清除速率记录
     */
    private void clearRateRecord() {
        synchronized (throughputMap) {
            for (var value : throughputMap.values()) {
                value.time = 0;
                value.value = 0;
            }
        }
    }

    /**
     * 线程调节器
     */
    private final Threads modulation = new Threads(1000) {
        private final LinkedList<Double> history = new LinkedList<>();// 历史速率
        private int smoothing;// 滤波次数
        private boolean increase = true;// 是否为增加状态
        private int now;// 当前次数
        private int immobility;// 速率持续不变的次数

        @Override
        public void run() {
            long time = System.currentTimeMillis();
            double speed = calcRate(time);
            setRate(speed * 1000d);
            while (history.size() > 12)
                history.removeLast();
            history.addFirst(speed);
            if (++now > smoothing)
                now = 0;
            else
                return;
            if (history.size() < smoothing + 2)
                return;
            double fluctuate = fluctuate();
            smoothing = N.toInt(fluctuate * 10d);
            // 计算速率比
            double right = rate(0, smoothing + 1);
            double left = rate(1, smoothing + 1);
            double rate = left == 0 || right == 0 ? 0 : right / left;
            // 调整线程
            int thread = modulation(rate);
            if (thread < 1)
                return;
            thread = Math.max(minThreadCount, thread);
            thread = Math.max(smoothing, thread);
            thread = Math.min(maxThreadCount, thread);
            pool.setMaxThreadCount(thread);
        }

        /**
         * 计算波动差值
         * @return 波动差值
         */
        private double fluctuate() {
            if (history.size() <= 2)
                return 0;
            var iterator = history.iterator();
            double left = iterator.next();
            double right = iterator.next();
            boolean increase = right > left;
            double all = right + left;
            double difference = 0;
            while (iterator.hasNext()) {
                right = iterator.next();
                all += right;
                boolean is = right > left;
                if (is != increase)
                    difference += Math.abs(right - left);
                left = right;
            }
            return all == 0 ? 0 : difference / all;
        }

        /**
         * 计算速率
         * @param start 开始位置
         * @param len 采样长度
         * @return 速率
         */
        private double rate(int start, int len) {
            int index = -1;
            double sum = 0;
            for (double v : history) {
                if (++index < start)
                    continue;
                sum += v;
                if (--len <= 0)
                    break;
            }
            return sum / Math.max(1, index - start + 1);
        }

        /**
         * 计算调节线程数
         * @param rate 速率比
         */
        private int modulation(double rate) {
            if (pool.getNowThreadCount() < pool.getMaxThreadCount())
                return -1;
            int thread = pool.getMaxThreadCount();
            // 增加阈值
            double add = 1.2;
            // 减少阈值
            double sub = 0.8;
            if (rate > sub && rate < add)
                if (++immobility < 5)
                    return -1;
            immobility = 0;
            increase = increase ? rate > add : rate < sub;
            int plus = Math.min(10, Math.max(1, thread >> 1));
            if (increase)
                thread += plus;
            else
                thread -= plus;
            return thread;
        }

        @Override
        public void onStop() {
            super.onStop();
            history.clear();
            increase = true;
            clearRateRecord();
        }
    };

    /**
     * 流量值记录器
     */
    private static class ThroughputRecord {
        public long time;// 记录时间戳
        public long value;// 记录值

        /**
         * 记录
         *
         * @param throughput 流量计数器
         * @param time       时间戳
         */
        public void record(Throughput throughput, long time) {
            this.time = time;
            value = throughput.size();
        }
    }
}
